package cn.itcast.up.bulkload;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2;
import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * 使用Bulkload将Hive中的数据,装载到HBase
 */
public class Hive2HBase {

    //1. 创建一个MR的Job对象
    //   1. 定义一个job的configuration对象.
    //2. 设置job的相关配置
    //   1. mapper class
    //      1. 编写map程序
    //         1. 读取Hive原始文件
    //         2. 将读取到的数据构建成key/value键值对(rowkey, put对象)
    //   2. map 输入的key
    //   3. map 输出的key
    //   4. jar的class
    //3. 利用HBase的Reduce实现生成HFile的功能.
    //4. 设置输入和输出路径信息
    //5. 执行job
    //6. 使用HBase的Bulkload功能,将HFile装载到HBase.

    //zk连接参数
    public static final String ZK_PARAM = "bd001:2181";
    // zk中HBase的节点路径
    public static final String ZK_NODE_PATH = "/hbase-unsecure";
    //hive表数据目录
    public static final String INPUT_PATH = "hdfs://bd001:8020/user/hive/warehouse/tags_dat.db/tbl_users";
    //生成的hfile目录 hdfs dfs -rmr /output_hfile/tbl_users
    public static final String HFILE_PATH = "hdfs://bd001:8020/output_hfile/tbl_users";
    //表名
    // create "tbl_test_users","detail" 新建表
    // truncate "tbl_test_users" 重建表
    // scan "tbl_test_users",{LIMIT=>3}
    public static final String TABLE_NAME = "tbl_test_users";
    // 列簇名字
    public static final String FAMILY_NAME = "detail";


    {
       new ArrayList<String>(){
           {
               add("hello");
               add("hello");
               add("hello");
               add("hello");
               add("hello");
               add("hello");
           }
       };
    }

    //表字段, new 类名(){}创建的是该类的子类对象,{{代码块,随着对象的创建而加载}}
    public static final List<String> fieldNameList = new ArrayList<String>() {{
        add("id");
        add("siteid");
        add("avatarimagefileid");
        add("email");
        add("username");
        add("password");
        add("salt");
        add("registertime");
        add("lastlogintime");
        add("lastloginip");
        add("memberrankid");
        add("bigcustomerid");
        add("lastaddressid");
        add("lastpaymentcode");
        add("gender");
        add("birthday");
        add("qq");
        add("job");
        add("mobile");
        add("politicalface");
        add("nationality");
        add("validatecode");
        add("pwderrcount");
        add("source");
        add("marriage");
        add("money");
        add("moneypwd");
        add("isemailverify");
        add("issmsverify");
        add("smsverifycode");
        add("emailverifycode");
        add("verifysendcoupon");
        add("canreceiveemail");
        add("modified");
        add("channelid");
        add("grade_id");
        add("nick_name");
        add("is_blacklist");
    }};

    public static void main(String[] args) throws Exception {
        //我们需要操作HBase,后面一定会用到Hbase的相关配置,我们可以直接使用Hbase的Configuration
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", ZK_PARAM);
        configuration.set("zookeeper.znode.parent", ZK_NODE_PATH);

        //1. 创建一个MR的Job对象
        Job job = Job.getInstance(configuration);
        //   1. mapper class
        job.setMapperClass(MyMapper.class);
        //   2. map 输出的key
        job.setMapOutputKeyClass(ImmutableBytesWritable.class);
        //   3. map 输出的value
        job.setMapOutputValueClass(Put.class);
        //   4. jar的class
        job.setJarByClass(Hive2HBase.class);

        // 利用HBase的Reduce实现生成HFile的功能.
        // 创建表
        Connection conn = ConnectionFactory.createConnection(configuration);
        Table table = conn.getTable(TableName.valueOf(TABLE_NAME));
        // 创建一个region的定位器
        RegionLocator regionLocator = conn.getRegionLocator(TableName.valueOf(TABLE_NAME));
        //生成HFile文件(Reduce)
        HFileOutputFormat2.configureIncrementalLoad(job, table, regionLocator);

        // 设置输入/输出路径
        FileInputFormat.setInputPaths(job, new Path(INPUT_PATH));
        FileOutputFormat.setOutputPath(job, new Path(HFILE_PATH));
        // 执行job
        job.waitForCompletion(true);
        //执行bulkload
        LoadIncrementalHFiles bulkload = new LoadIncrementalHFiles(configuration);
        bulkload.doBulkLoad(
                new Path(HFILE_PATH),
                conn.getAdmin(),
                table,
                regionLocator
        );
        System.out.println("Bulkload执行完毕");

    }
}

/**
 * 自定义Mapper,将Hive中的数据读取出来,转换为rowkey和put
 * <LongWritable, Text, ImmutableBytesWritable, Put>
 *     Hive文件的行号, Hive的文本内容, 输出的rowKey, 输出的Put.
 */
class MyMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, Put> {
    @Override
    protected void map(LongWritable lineNumber, Text rowLine, Context context) throws IOException, InterruptedException {
        // 1. 将rowLine中的数据进行切割."\001"
        String[] tempArr = rowLine.toString().split("\001");
        // 为了方便后面的操作,可以将数组转换为集合.
        List<String> valueList = Arrays.asList(tempArr);

        // 因为进行split切割的时候,如果最后的字段值为空,那么集合的个数会变少
        // 我们可以判断一下缺失了几个字段,然后对缺失的字段进行补充.
        int num = Hive2HBase.fieldNameList.size() - valueList.size();
        //开始补充num个字段
        for (int i = 0; i < num; i++) {
            valueList.add("");
        }

        // 2. 使用列名/列值进行数据合并,构建Hbase中的每一行数据.
        // 构建put对象,使用第一列作为rowkey.
        Put put = new Put(valueList.get(0).getBytes());
        for (int i = 0; i < valueList.size(); i ++ ){
            put.addColumn(
                    Hive2HBase.FAMILY_NAME.getBytes(),
                    Hive2HBase.fieldNameList.get(i).getBytes(),
                    valueList.get(i).getBytes()
            );
        }
        // 返回数据
        context.write(new ImmutableBytesWritable(valueList.get(0).getBytes()), put);
    }
}
